package scala.actors.multicast.actorgroup

import scala.actors.Actor.loop
import scala.actors.Actor.actor
import scala.actors.Actor.self
import scala.actors.multicast.message.ActorMessage
import scala.actors.multicast.message.ActorCommand
import scala.actors.multicast.message.Bang
import scala.actors.multicast.message.BangMark
import scala.actors.multicast.message.BangBang
import scala.actors.multicast.message.Forward
import scala.collection.mutable.HashSet
import scala.collection.mutable.SynchronizedSet

/* An ActorGroupCommand is used to encode control messages for ActorGroups.
   The only such message right now is Shutdown, which cleans up any resources
   and then terminates the actor. */
abstract class ActorGroupCommand
case object Shutdown extends ActorGroupCommand

/* An ActorGroup tracks a set of Actors. Messages sent to the group are relayed
   to all actors in the group. Message replies are collated and returned as a
   Set of Any. */
class ActorGroup extends Actor {
  
  /* Start the ActorGroup; an ActorGroup is an Actor, thus it will not behave
     properly unless its start method is called. */
  start
  
  /* The set of actors belonging to this group. It is a SynchronizedSet because
     any thread may add an actor to the group at any time. */
  private[actorgroup] val actors: SynchronizedSet[Actor] = new HashSet[Actor] with SynchronizedSet[Actor]
  
  /* Add the given actor to this group. The group cannot be added to itself, or
     else an infinite loop ensues. */
  def +=(actor: Actor): Unit = if (actor != this) actors += actor
  
  /* Remove the given actor from this group. */
  def -=(actor: Actor): Unit = actors -= actor
  
  /* Returns a string representation of the set of actors in this group. */
  override def toString: String = actors.toString
  
  /* These methods intercept the calls to their respective super implemen-
     tations and pack the message into a tuple. This allows us to later
     determine which method was called. */
  override def !(msg: Any): Unit = super.! (pack(Bang, msg))
  override def forward(msg: Any): Unit = super.forward (pack(Forward, msg))
  override def !?(msg: Any): Any = super.!? (pack(BangMark, msg))
  override def !?(msec: long, msg: Any): Option[Any] = super.!? (msec, pack(BangMark, msg))
  override def !!(msg: Any): Future[Any] = super.!! (pack(BangBang, msg))
  override def !![a](msg: Any, f: PartialFunction[Any, a]): Future[a] = super.!! (pack(BangBang, msg), f)
  
  /* Pack the given command and message into a tuple. If the message is already
     a tuple, pass it through unchanged. Ignore Shutdown messages. This is
     necessary because !? and !! are implemented in terms of !, and dynamic
     dispatch causes our locally overridden methods to be called. */
  private def pack(cmd: ActorCommand, msg: Any): Any = msg match {
    case (Bang, _) => msg
    case (Forward, _) => msg
    case (BangMark, _) => msg
    case (BangBang, _) => msg
    case Shutdown => msg
    case _ => (cmd, msg)
  }
  
  def act: Unit = {
    loop {
      receive {
        /* Do (actor ! msg) for each actor in the group. */
        case (Bang, msg) => actors.foreach(actor => if (actor != sender) actor ! msg)
        
        /* Do (actor forward msg) for each actor in the group. */
        case (Forward, msg) => actors.foreach(actor => if (actor != sender) actor forward msg)
        
        /* Do (actor !? msg) for each actor in the group and reply with the set
           of results. */
        case (BangMark, msg) => {
          var responses = new HashSet[Any]
          actor {
            actors -= self
            responses.synchronized {
              actors.foreach(actor => if (actor != sender) responses += (actor !? msg))
              /* Signal that the responses are ready. */
              responses.notifyAll
            }
          }
          responses.synchronized {
            /* Wait for the responses to be produced. */
            responses.wait
            reply(responses)
          }
        }
        
        /* Do (actor !! msg) for each actor in the group and reply with the set
           of results. */
        case (BangBang, msg) => {
          val responses = new HashSet[Any]
          actor {
            actors -= self
            responses.synchronized {
              actors.foreach(actor => if (actor != sender) responses += (actor !! msg))
              /* Signal that the responses are ready. */
              responses.notifyAll
            }
          }
          responses.synchronized {
            /* Wait for the responses to be produced. */
            responses.wait
            reply(responses)
          }
        }
        
        /* No cleanup is necessary upon shutdown. */
        case Shutdown => exit
      }
    }
  }
}
